package file_text_input;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

public class MyRecordReader extends RecordReader<LongWritable, Text> {
//    private static final int BUFFER_SIZE = 5;
//    private static final String DELIMITER = ":";

//    private Queue<String> valueBuffer = new LinkedList<String>();
//    private Queue<Long> keyBuffer = new LinkedList<Long>();
    //flag para pular a linha que tem o tamanho do bloco;
    private boolean flagBlockSize = true;
    private long start = 0;
	private long end;
	private int current = 0;
	private LongWritable currentKey;
	private Text currentValue;
    private static String SOURCE = "owl.source";
	private static String TARGET = "owl.target";
	ArrayList<Integer> indexPairSource;
	ArrayList<Integer> indexPairTarget;
	private int currentSourceIndex;
	private int endSourceIndex;
	private int currentTargetIndex;
	private int endTargetIndex;
	private boolean isSource = false;
	private boolean isTarget = false;

    private RecordReader<LongWritable, Text> rr;
    public MyRecordReader(InputSplit split, Configuration conf, RecordReader<LongWritable, Text> rr) {
        this.rr = rr;
		
        List<String> ontologySource = Arrays.asList(conf.getStrings(SOURCE));
		List<String> ontologyTarget = Arrays.asList(conf.getStrings(TARGET));
		
		String[] pathFile = split.toString().split("/");
		String nameFile = pathFile[pathFile.length - 1].split(":")[0];
		
		indexPairSource = getArrayIndex(ontologySource, nameFile);			
		if (!indexPairSource.isEmpty()) {
			isSource = true;
		}
		
		indexPairTarget = getArrayIndex(ontologyTarget, nameFile);
		if (!indexPairTarget.isEmpty()) {
			isTarget = true;
		}
		
		currentSourceIndex = 0;
		endSourceIndex = indexPairSource.size();
		currentTargetIndex = 0;
		endTargetIndex = indexPairTarget.size();
    }

    @Override
    public void close() throws IOException {
        rr.close();
    }

    @Override
    public LongWritable getCurrentKey() throws IOException, InterruptedException {
    	
    	return currentKey;
//        return key;
    }

    @Override
    public Text getCurrentValue() throws IOException, InterruptedException {
    	return currentValue;
//        return value;
    }

    @Override
    public float getProgress() throws IOException, InterruptedException {
        return rr.getProgress();
    }

    @Override
    public void initialize(InputSplit arg0, TaskAttemptContext arg1)
            throws IOException, InterruptedException {
        rr.initialize(arg0, arg1);
    }

    @Override
    public boolean nextKeyValue() throws IOException, InterruptedException {
    	rr.nextKeyValue();
    	currentKey = new LongWritable(current);
    	currentValue = rr.getCurrentValue();
		
    	if (flagBlockSize) {
			flagBlockSize = false;
			end = Long.parseLong(currentValue.toString()) - 1;
			rr.nextKeyValue();
			currentKey = new LongWritable(current);
			currentValue = rr.getCurrentValue();
		}
		return next();
//        if (valueBuffer.isEmpty()) {
//            while (valueBuffer.size() < BUFFER_SIZE) {
//                if (rr.nextKeyValue()) {
//                    keyBuffer.add(rr.getCurrentKey().get());
//                    valueBuffer.add(rr.getCurrentValue().toString());
//                } else {
//                    return false;
//                }
//            }
//        } else {
//            if (rr.nextKeyValue()) {
//                keyBuffer.add(rr.getCurrentKey().get());
//                valueBuffer.add(rr.getCurrentValue().toString());
//                keyBuffer.remove();
//                valueBuffer.remove();
//            } else {
//                return false;
//            }
//        }
//        key.set(keyBuffer.peek());
//        value.set(getValue());
//        return true;
    }
    
    private ArrayList<Integer> getArrayIndex(List<String> ontologySource, String nameFile) {
		ArrayList<Integer> out = new ArrayList<Integer>();
		for (int i = 0; i < ontologySource.size(); i++) {
			if (nameFile.equals(ontologySource.get(i))) {
				out.add(i);
			}
		}
			
		return out;
	}
    
    private boolean next() throws IOException, InterruptedException {
    	if (current < end) {
			boolean isSource = false;
			String[] inputTXT = currentValue.toString().split("\\<\\*>");
			if (currentSourceIndex < endSourceIndex) {
				currentValue.set(indexPairSource.get(currentSourceIndex) + "." + current + "." + 1 + "." + inputTXT[0] + "." + inputTXT[2]);
				currentSourceIndex++;
				isSource = true;
			} else {
				if (currentTargetIndex < endTargetIndex) {
					currentValue.set(indexPairSource.get(currentSourceIndex) + "." + current + "." + 2 + "." + inputTXT[0] + "." + inputTXT[2]);
					currentTargetIndex++;
				}
			}
			
			if (currentSourceIndex == endSourceIndex && currentTargetIndex == endTargetIndex) {
				currentSourceIndex = 0;
				currentTargetIndex = 0;
				current++;
				//return rr.nextKeyValue();
			}
			
			return true;
		}
		return false;
	}
//    private String getValue() {
//        StringBuilder sb = new StringBuilder();
//        Iterator<String> iter = valueBuffer.iterator();
//        while (iter.hasNext()) {
//            sb.append(iter.next());
//            if (iter.hasNext()) sb.append(DELIMITER);
//        }
//        return sb.toString();
//    }

}
